package org.geektimes.reactive.streams;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

/**
 * Created with IntelliJ IDEA.
 * Description:
 *
 * @Author: menglinggang
 * @Date: 2021-03-31
 * @Time: 3:20 下午
 */
public class DefaultSubscriber<T> implements Subscriber<T> {

    private Subscription subscription;

    private int count;

    @Override
    public void onSubscribe(Subscription subscription) {
        if (this.subscription != null) {
            this.onError(new RuntimeException("subscriber only can subscribe once"));
        }
        this.subscription = subscription;
    }

    @Override
    public void onNext(T t) {

        if (++count == 2) { // 当到达数据阈值时，取消 Publisher 给当前 Subscriber 发送数据
            subscription.cancel();
        }
        System.out.println("收到数据：" + t);

    }

    @Override
    public void onError(Throwable throwable) {
        System.out.println("遇到异常：" + throwable);
    }

    @Override
    public void onComplete() {
        System.out.println("收到数据完成");
    }
}
